package com.sqllog.admin.init;

import cn.hutool.core.collection.CollectionUtil;
import com.sqllog.admin.cache.CanalClientCache;
import com.sqllog.admin.sql.entity.CanalClientConfigInfo;
import com.sqllog.admin.sql.entity.LogDatabaseManager;
import com.sqllog.admin.sql.listener.DataBaseLogListener;
import com.sqllog.admin.sql.mapper.CanalClientConfigInfoMapper;
import com.sqllog.admin.sql.mapper.LogDatabaseManagerMapper;
import com.sqllog.listener.SimpleCanalEventListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * TODO
 *
 * @author chennengcheng
 * @date 2022-01-07 17:19
 * @remark
 */

@Slf4j
@Component
@Order(value = 0)
public class CanalClientInit implements CommandLineRunner {

    @Resource
    private CanalClientConfigInfoMapper canalClientConfigInfoMapper;

    @Resource
    private LogDatabaseManagerMapper logDatabaseManagerMapper;

    @Override
    public void run(String... args) {
        // 初始化表
        List<LogDatabaseManager> logDatabaseManagers = logDatabaseManagerMapper.selectAll();
        if (CollectionUtil.isNotEmpty(logDatabaseManagers)) {
            Map<String, List<LogDatabaseManager>> logDatabaseMap = logDatabaseManagers.stream().collect(Collectors.groupingBy(LogDatabaseManager::getQueryName));
            CanalClientCache.logDatabaseManageMap.putAll(logDatabaseMap);
        }

        //初始化 客户端
        List<CanalClientConfigInfo> canalClientConfigInfos = canalClientConfigInfoMapper.selectByAll();
        if (CollectionUtil.isEmpty(canalClientConfigInfos)) {
            return;
        }

        for (CanalClientConfigInfo canalClientConfigInfo : canalClientConfigInfos) {
            log.info("启动 " + canalClientConfigInfo.getInstanceName());
            SimpleCanalEventListener simpleCanalEventListener = new DataBaseLogListener();
            List<SimpleCanalEventListener> simpleCanalEventListeners = new ArrayList<>();
            simpleCanalEventListeners.add(simpleCanalEventListener);
            CanalClientConfigInfo.startCanalClient(canalClientConfigInfo, simpleCanalEventListeners);
        }



    }
}
